<?php

declare(strict_types=1);

namespace App\Amqp\Consumer;

use Hyperf\Amqp\Annotation\Consumer;
use Hyperf\Amqp\Message\ConsumerMessage;
use Hyperf\Amqp\Result;
use PhpAmqpLib\Message\AMQPMessage;
use Hyperf\Amqp\Message\Type;
use Service\AppService\MonitorService;
use Service\ConsumerService\SkuChangeLogMessageService;

/**
 * qos (prefetch_size,prefetch_count,global)
 * @Consumer(exchange="COMMODITY_SKU_CHANGE_LOG_MESSAGE", routingKey="COMMODITY_SKU_CHANGE_LOG_MESSAGE", queue="COMMODITY_SKU_CHANGE_LOG_MESSAGE", nums=1)
 */
class SkuUserChangeLogMessageConsumer extends ConsumerMessage
{
    protected $type = Type::DIRECT;//直连模式
    protected $qos  = [
        'prefetch_count'    => 10
    ];
    public function consumeMessage($data, AMQPMessage $message): string
    {
        try
        {
            SkuChangeLogMessageService::instance()->syncChangeAddLog($data);
            return Result::ACK;
        }
        catch ( \Throwable $t )
        {
            $error = [
                'error'         => $t->getMessage(),
                'errorFile'     => $t->getFile(),
                'errorLine'     => $t->getLine(),
            ];
            
            $contents = json_encode($error,JSON_UNESCAPED_UNICODE);
            MonitorService::instance()->sendMsg($contents);
            sleep(10);//休眠10秒之后，重回队列
            return Result::REQUEUE;
        }
    }
    
    /**
     * 设置是否自启动
     */
    public function isEnable() : bool
    {
        $isDisAable = env('AMQP_COMMODITY_SKU_CHANGE_LOG_MESSAGE_DISABLE');
        return $isDisAable ? false : true ;
    }
}